package com.yanggu.flink.datastream_api.window

import com.yanggu.flink.datastream_api.pojo.SensorReading
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

import java.time.Duration

/**
 * 窗口1 [195, 210)
 * 窗口2 [210, 225)
 * 窗口3 [225, 240)
 * 窗口4 [240, 255)
 * 窗口5 [255, 270)
 * 窗口6 [270, 285)
 * 窗口7 [285, 300)
 */
/**
 * https://www.cnblogs.com/rossiXYZ/p/12286407.html
 * 聚合类的处理
 * Flink可以每来一个消息就处理一次，但是有时我们需要做一些聚合类的处理，例如：在过去的1分钟内有多少用户点击了我们的网页。所以Flink引入了窗口概念。
 * 窗口 窗口的作用为了周期性的获取数据。就是把传入的原始数据流切分成多个buckets，所有计算都在单一的buckets中进行。窗口（window）就是从 Streaming 到 Batch 的一个桥梁。
 * 带来的问题：聚合类处理带来了新的问题，比如乱序/延迟。其解决方案就是 Watermark / allowedLateNess / sideOutPutLateData 这一组合拳。
 * Watermark 是防止数据乱序 / 本质上是时间戳, 定义了什么时候不再等待数据, 触发窗口进行计算。Watermark时间戳 = MAX(EventTime) - 延迟时间 - 1ms。
 * 窗口计算：凡是窗口的end_time <= Watermark时间戳, 都会触发窗口计算。同时Flink将窗口触发计算和关闭进行了分离
 * allowedLateNess 本来窗口触发计算后就应该关闭, 但是如果还有迟到的数据来了怎么办呢？allowedLateNess是将窗口关闭时间再延迟一段时间，在此期间来的数据都会触发窗口重新进行计算。
 * 当某个数据的来临时, 且窗口的end_time <= MAX(EventTime) - Watermark时间戳 - allowedLateNess延迟时间, 窗口会被关闭。
 * sideOutPutLateData 当窗口被关闭后, 迟到的数据会直接丢弃。当数据所属的属于窗口已经彻底关闭后，如果设定了迟到数据的侧输出流，就会把所有过期延迟数据放到侧输出流，让用户决定如何处理，是最后兜底操作。
 * 总结起来就是说
 * Windows -----> Watermark -----> allowedLateNess -----> sideOutPutLateData
 * 用Windows把流数据分块处理，用Watermark确定什么时候不再等待更早的数据/触发窗口进行计算，用allowedLateNess将窗口关闭时间再延迟一段时间。用sideOutPutLateData最后兜底把数据导出到其他地方。
 */
object WindowDemo02 {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //从flink1.12开始默认的时间处理方式为eventTime, 无需显示设定
    //env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.setParallelism(1)
    //设置Watermark的生成时间间隔
    //env.getConfig.setAutoWatermarkInterval(50L)

    val lateTag = new OutputTag[SensorReading]("late")

    //每15秒统计一次, 窗口内各传感器所有温度的最小值, 以及最新的时间戳
    val stream = env.socketTextStream("localhost", 9000)
      .name("socket data source").uid("socket data source")
      //将字符串按照","切分后, 转换成scala的样例类
      .map(value => {
        val arr = value.split(",")
        SensorReading(arr(0), arr(1).toLong * 1000, arr(2).toDouble)
      })
      .name("split map").uid("split map")
      //定义EventTime(TimestampAssigner)和Watermark(WatermarkGenerator 3秒钟的延迟)
      .assignTimestampsAndWatermarks(WatermarkStrategy
        //静态泛型可以写在方法名后面
        .forBoundedOutOfOrderness[SensorReading](Duration.ofSeconds(3L))
        //scala的匿名函数如何序列化
        //.withTimestampAssigner((element, _) => element.timestamp * 1000)
        .withTimestampAssigner(new SerializableTimestampAssigner[SensorReading] {
          override def extractTimestamp(element: SensorReading, recordTimestamp: Long): Long = element.timestamp
        })
      )
      //根据id进行分组 DataStream => KeyedStream
      .keyBy(_.id)
      //定义了一个事件时间滚动窗口, 窗口大小为15秒
      .window(TumblingEventTimeWindows.of(Time.seconds(15L)))
      //允许数据最大延迟60秒
      .allowedLateness(Time.seconds(60L))
      //将迟到的数据打上一个tag, 以便后续的处理获取
      .sideOutputLateData(lateTag)
      //聚合运算, 获取最小温度和最新时间戳
      .reduce((curRes, newData) => SensorReading(curRes.id, newData.timestamp, curRes.temperature.min(newData.temperature)))
      .name("minTemperature newest timestamp").uid("minTemperature newest timestamp")

    //将聚合后的窗口数据进行打印输出
    stream.print("result")
    //通过lateTag获取迟到的数据, 然后打印输出
    stream.getSideOutput(lateTag).print("late")

    //执行job
    env.execute("window test")
  }

}
